package org.kth.dks.dks_marshal;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.log4j.Logger;
import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSNetAddress;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_exceptions.DKSNodeAlreadyRegistered;
import org.kth.dks.util.DKSPrintTypes;

public class DKSMarshal {

  private Logger log = Logger.getLogger(DKSMarshal.class);
  private static final String BASEPACKAGE = "org.kth.dks.";
  private boolean LOGICAL_COMM = false;
  private ConnectionManager cm = null;
  private Map objectAdapterMap = new HashMap(); // DKSNetAddr->ObjectAdapter
  private Map objectAdapterThreadList = new HashMap(); // ObjectAdapter->Thread

  public static final byte TRANSTYPEXML = 0;
  public static final byte TRANSTYPEBINARY = 1;
  public static final byte TRANSDEFAULT = TRANSTYPEBINARY;

  public DKSMarshal(ConnectionManager cm) { //New
    this.cm = cm;
  }

  /**
   * Calls the unmarshaller and unmarshalls the byte stream to DKSMessage, and
   * dispatches it to the right handler
   * @param input byte[] raw byte array representing a full XML document
   * @param source DKSRef reference of the node that sent to message. The reference
   * is passed to the handler method when dispatching the event
   */
  public boolean unmarshalDispatch(byte transType, byte[] input,
                                   DKSNetAddress source) {
    DKSMessage dksMsg = DKSMessage.unmarshal(transType, input);
    return dksMsg == null ? false :
        dispatchOA(dksMsg, dksMsg.getSendRef(),
                   dksMsg.getRecvRef().getOverlayAddress());
  }

  /**
   * Takes an unmarshalled message from <i>source</i> to <i>dest</i> and dispatches it to the correct
   * ObjectAdapter
   * @param dksMsg DKSMessage already marshalled message
   * @param source DKSRef source node
   * @param dest DKSOverlayAddress destination node
   */
  public boolean dispatchOA(DKSMessage dksMsg, DKSRef source,
                            DKSOverlayAddress dest) {
    if (!objectAdapterMap.containsKey(dest)) {
      log.error(
          "Destination DKSOverlayAddress " + dest + " is not registered in the marshaler" + source);
      return false;
    }
    else {
//      DKSPrint.println(DKSPrintTypes.MSGS,
//                       "receive remote(" + dksMsg.getRecvRef().getID() + " from " + dksMsg.getSendRef().getID() +
//                       "): " + dksMsg.getName());
      ObjectAdapter oa = (ObjectAdapter) objectAdapterMap.get(dest);
      return oa.putMessagePair(dksMsg, source);
    }
  }

  /**
   * Takes an unmarshalled message from <i>source</i> to <i>dest</i> and dispatches it to the correct
   * ObjectAdapter
   * @param dksMsg MsgSrcDestWrapper triple wrapper containing DKSMessage, Source, Destination
   * @param dest DKSOverlayAddress destination node
   */
  public void failureHandler(MsgSrcDestWrapper triple) {
    if (!objectAdapterMap.containsKey(triple.getSrc().getOverlayAddress())) {
      log.error( "Source DKSOverlayAddress is not registered in the marshaler (in failureHandler)");
    }
    else {
      log.error(
                       "DKSMarshal -- Failure detected to" +
                       triple.getDest().getOverlayAddress());
      ObjectAdapter oa = (ObjectAdapter) objectAdapterMap.get(triple.getSrc().
          getOverlayAddress());
      oa.putMessagePair(new FailureMsg(triple.getSrc(), triple.getDest(),
                                       triple.getMsg()), null);
    }
  }

  public void registerDKSNode(DKSOverlayAddress addr) throws
      DKSNodeAlreadyRegistered {
    if (objectAdapterMap.containsKey(addr)) {
      throw new DKSNodeAlreadyRegistered();
    }

    ObjectAdapter oa = new ObjectAdapter();
    Thread oaThread = new Thread(oa);
    oaThread.setName(ObjectAdapter.class.getName());
    oaThread.start();
    objectAdapterThreadList.put(oa, oaThread);
    objectAdapterMap.put(addr, oa);
  }

  public void unregisterDKSNode(DKSOverlayAddress addr) {
    ObjectAdapter oa = (ObjectAdapter) objectAdapterMap.remove(addr);
    if (oa != null) {
      Thread oaThr = (Thread) objectAdapterThreadList.remove(oa);
      if (oaThr != null) {
        oaThr.interrupt();
      }
    }
  }

  /**
   * @see ObjectAdapter.addMsgHandler()
   * All class names are rooted at this.BASEPACKAGE
   */
  public boolean addMsgHandlerPrefixed(DKSOverlayAddress addr,
                                       String messageClassZ,
                                       String handlerClassZ,
                                       String handlerMethodZ,
                                       Object handlerObject) {
    return addMsgHandler(addr,
                         BASEPACKAGE + messageClassZ,
                         BASEPACKAGE + handlerClassZ,
                         handlerMethodZ,
                         handlerObject);
  }

  /**
   * @see ObjectAdapter.addMsgHandler()
   */
  public boolean addMsgHandler(DKSOverlayAddress addr,
                               String messageClassZ,
                               String handlerClassZ,
                               String handlerMethodZ,
                               Object handlerObject) {
    if (!objectAdapterMap.containsKey(addr)) {
      log.warn(
                     "Could not regisiter handler for " + handlerMethodZ
                     + ", because no ObjectAdapter is registered yet\n");
      return false;
    }
    else {
      ObjectAdapter oa = (ObjectAdapter) objectAdapterMap.get(addr);
      return oa.addMsgHandler(messageClassZ,
                              handlerClassZ,
                              handlerMethodZ,
                              handlerObject);
    }
  }

  /**
   * Sends the message inside msg from src to dest. If the destination  is
   * local it is just given to that node without any marshaling/unmarshaling. If the destination
   * is remote, the message is unmarshalled before being sent.
   * @param src DKSRef
   * @param dest DKSRef
   * @param msg DKSMessage
   * @return boolean true if successful
   */

  public boolean send(DKSRef src, DKSRef dest, DKSMessage msg) {

    if (src.getIP().equals(dest.getIP()) && src.getPort() == dest.getPort() &&
        LOGICAL_COMM) {
      log.info(
                       "send local(" + src.getID() + " to " + dest.getID() +
                       "): " + msg.getName());
      log.info(
                       "receive local(" + dest.getID() + " from " + src.getID() +
                       "): " + msg.getName());
      dispatchOA(msg, src, dest.getOverlayAddress());
      return true;
    }
    else {
    	log.info(
                       "send remote(" + dest.getID() + " to " + src.getID() +
                       "): " + msg.getName());

      return cm.send(new MsgSrcDestWrapper(msg, src, dest));
    }
  }

  public byte[] marshalMsgSrcDestWrapper(MsgSrcDestWrapper wr) {
    final DKSMessage msg = wr.getMsg();
    final DKSRef src = wr.getSrc();
    final DKSRef dest = wr.getDest();
    msg.setRecvRef(dest);
    msg.setSendRef(src);

    return msg.flatten();
  }

  /**
   * Cleans the object adapter map such that the object adapters and their corresponding buffers can be GC:d
   */
  public void end() {
    objectAdapterMap.clear();
    for (Iterator iter = objectAdapterThreadList.values().iterator();
         iter.hasNext(); ) {
      Thread oaThread = (Thread) iter.next();
      oaThread.interrupt();
    }
    objectAdapterThreadList.clear();
  }

} //DKSMarshal class
